using System;
using System.Data;
using System.ServiceModel;
using System.Collections.Generic;
using System.Collections;
using System.Runtime.Serialization;
using System.ServiceModel.Channels;
using System.ServiceModel.Description;
using System.Diagnostics;
using System.Threading;
using ChunkTransporter.ChunkTransporterService;

/****
 * 
 * Server side will keep the table for the life time of the transport
 * Service creates cookie (unique table id), 
 * Service returns TableManifest to client: TableId, size of payload, number of records
 * 
 * Client creates wrapper, which extracts information needed to transfer table from service
 * client creates a transport plan:  threads, etc
 * client executes threads, receives chunks from service, waits, merges results
 * 
 * server can free resources

 * 
 *	***	Server	*** 
 * 
 * using System.Data;
 * using ChunkTransporter;
 * using System.ServiceModel;
 * 
 * namespace WCFService
 * {
 *     [ServiceContract()]
 *     public interface IService1
 *     {
 *         [OperationContract]
 *         TableManifest GetCustomers();
 *     }
 * 
 *     [ServiceBehavior]
 *     public class MyService : IService1
 *     {
 *         public TableManifest GetCustomers()
 *         {
 *             DataTable dtCust = DBAdapter.Fill("Select * from Customers");***	Client	***
 *             return new TableManifest(ref dtCust);
 *         }
 *         ...
 *     }
 * }
 * 
 * 
 *	***	Client	*** 
 * 
 * using System.Data;
 * using ChunkTransporter;
 * using System.ServiceModel;
 * namespace WCFClient
 * {
 * 	class TheClient
 * 	{
 * 		DataTable Customers
 * 		{
 * 			get
 * 			{
 * 				Service1Client theService = new Service1Client();
 * 				TableManifest tableManifest = theService.GetCustomers();
 * 				DataTable dtCust = TableManifest.FetchTable(tableManifest);
 * 				return dtCust;				
 * 			}
 * 		}
 * 		...
 * 	}
 * }
 *
 ****/


namespace ChunkTransporter
{

	[ServiceContract()]
	public interface ITableTransporterService
	{
		[OperationContract]
		DataTable NextChunk(int tableId);
	}

	#region TableManifest
	/// <summary>
	/// This is what gets returned from the service.
	/// </summary>
	[DataContract]
	public class TableManifest
	{
		#region private data members
		DataTable _schema;
		int _numRecords;
		int _tableSizeBytes;
		string _baseAddresses;
		int _id;
		#endregion


		/// <summary>
		/// Service should consider making a copy of the table, 
		/// as it will get modified in transit.
		/// </summary>
		/// <param name="dt"></param>
		public TableManifest(ref DataTable dt)
		{
			_schema = dt.Clone();
			_numRecords = dt.Rows.Count;
			_tableSizeBytes = DataTableHelper.InterpolateTableSize(dt);
			_id = TableWearhouse.Instance.Add(new Table(ref dt));
			_baseAddresses = TableWearhouse.URL;
		}

		/// <summary>
		/// Client calls this method, in order to harvest the DataTable
		/// This call goes to fetch the data table, returning once all chunks are in place.
		/// </summary>
		/// <param name="tableManifest">As returned by the service</param>
		/// <returns>The large data table</returns>
		public static DataTable FetchTable(object tableManifest)
		{

			TableManifest tm = new TableManifest((object)tableManifest);
			if (tm == null)
				throw new ArgumentException("GetTable(TableManifest) TableManifest invalid");
			try
			{
				return new TableTop().GetTable(tm);
			}
			catch (Exception e)
			{
				Helper.LogException("GetTable()", e);
				throw;
			}
		}

		/// <summary>
		/// This constructor is used internally, 
		/// when translating a TableManifest as returned from the local service reference wrapper
		/// </summary>
		/// <param name="other"></param>
		private TableManifest(object other)
		{
			Type manifestType = other.GetType();
			if (manifestType.GetProperty("Schema") == null)
				throw new ArgumentException("Bad TableManifest");

			_id = (int)manifestType.GetProperty("Id").GetGetMethod().Invoke(other, null);
			_schema = manifestType.GetProperty("Schema").GetGetMethod().Invoke(other, null) as DataTable;
			_numRecords = (int)manifestType.GetProperty("NumRecords").GetGetMethod().Invoke(other, null);
			_tableSizeBytes = (int)manifestType.GetProperty("SizeBytes").GetGetMethod().Invoke(other, null);
			_baseAddresses = (string)manifestType.GetProperty("BaseAddresses").GetGetMethod().Invoke(other, null);
		}

		[DataMember]
		public string BaseAddresses
		{
			get { return _baseAddresses; }
			set { _baseAddresses = value; }
		}
		[DataMember]
		public int NumRecords
		{
			get { return _numRecords; }
			set { _numRecords = value; }
		}
		[DataMember]
		public int SizeBytes
		{
			get { return _tableSizeBytes; }
			set { _tableSizeBytes = value; }
		}
		[DataMember]
		public int Id
		{
			get { return _id; }
			set { _id = value; }
		}
		[DataMember]
		public DataTable Schema
		{
			get { return _schema; }
			set { _schema = value; }
		}
	}
	#endregion

	#region Client TableTop

	/// <summary>
	/// Create several threads, each thread makes multiple requests to the service.
	/// </summary>
	internal class TableTop
	{
		TableManifest _tableManifest;
		DataTable _theTable;
		List<IAsyncResult> _results = new List<IAsyncResult>();
		bool _isFetchComplete = false;

		internal DataTable GetTable(TableManifest tm)
		{
			_tableManifest = tm;
			_theTable = tm.Schema;
			// to speed up, remove the PK, and beginLoadData
			DataColumn[] primaryKey = _theTable.PrimaryKey;
			_theTable.PrimaryKey = null;
			_theTable.BeginLoadData();

			SpawnThreads();
			WaitForThreads();

			//reinstate primary key
			_theTable.EndLoadData();
			_theTable.PrimaryKey = primaryKey;

			return _theTable;
		}

		delegate void AsyncRequestDelegate();
		void workerThread()
		{
			if (_isFetchComplete)
				return;
			if (Thread.CurrentThread.Name == null)
				Thread.CurrentThread.Name = "WCF ChunkTransporter worker thread";

			ChunkTransporterService.TableTransporterServiceClient service = null;
			try
			{
				service = WireupClientsideEndpoints();

				DataTable currentChunk = null;
				try
				{
					service = WireupClientsideEndpoints();
					while (!_isFetchComplete)
					{
						currentChunk = service.NextChunk(_tableManifest.Id);
						if (currentChunk != null && currentChunk.Rows.Count > 0)
						lock (_theTable)
							foreach (DataRow dr in currentChunk.Rows)
								_theTable.Rows.Add(dr.ItemArray);
						else
							_isFetchComplete = true;
					}


					Helper.TRACE("CT {0}:{1} records worker TID: {2} exiting", _theTable.TableName, Helper.FormatLong(_theTable.Rows.Count), Thread.CurrentThread.ManagedThreadId);
				}
				catch (Exception e)
				{
					Helper.LogException("NextChunk", e);
					return;
				}
			}
			finally
			{
				service.Close();
			}
		}

		void SpawnThreads()
		{
			AsyncRequestDelegate worker = new AsyncRequestDelegate(this.workerThread);
			for(int i=0; i< Config.ThreadCount; i++)
				_results.Add(worker.BeginInvoke(null, null));
		}

		void WaitForThreads()
		{
			foreach (IAsyncResult res in _results)
				if (!res.AsyncWaitHandle.WaitOne(TimeSpan.FromSeconds(10 * Config.RemotingTimeout), false))
					throw new TimeoutException("ChunkTransporter call has timed out");
		}

		private TableTransporterServiceClient WireupClientsideEndpoints()
		{
			string url = _tableManifest.BaseAddresses;
			//Helper.TRACE("Connecting with {0}", url);
			Debug.Assert(url != null, "URL=NULL");
			//if (_TableWearhouseService != null 
			//    && _TableWearhouseService.State == CommunicationState.Opened
			//    && _TableWearhouseService.Endpoint.ListenUri.OriginalString == url)
			//    return;

			//if (_TableWearhouseService != null)
			//    _TableWearhouseService.Close();

			NetTcpBinding binding = new NetTcpBinding();
			binding.CloseTimeout = binding.ReceiveTimeout = binding.SendTimeout = binding.OpenTimeout = TimeSpan.FromMinutes(10);
			binding.MaxReceivedMessageSize = 100000000;

			return new ChunkTransporterService.TableTransporterServiceClient
				(binding, new EndpointAddress(url));
		}
	}
	#endregion

	#region Server Table object
	/// <summary>
	/// Table object - lives at server.
	/// wraps the DataTable
	/// services chunk requests
	/// </summary>
	internal class Table
	{
		private DataTable _theBigTable;
		int _lastRequestOffset = 0;
		int _id;
		int _size;
		DateTime _constructionTime;

		internal string TableName { get { return _theBigTable.TableName; } }
		internal int Id { get { return _id; } set { _id = value; } }
		internal Table(ref DataTable t)
		{
			_theBigTable = t;
			_constructionTime = DateTime.Now;
			_theBigTable.RemotingFormat = SerializationFormat.Binary;
			_size = DataTableHelper.InterpolateTableSize(t);
		}
		internal int SizeBytes { get { return _size; } }
		internal DateTime ConstructionTime { get { return _constructionTime; } }


		internal DataTable GetChunk()
		{// service next chunk

			DataTable chunk = _theBigTable.Clone();
			chunk.PrimaryKey = null;
			chunk.Constraints.Clear();
			chunk.BeginLoadData();

			int rowStartIndex, rowEndIndex;
			lock (this)
			{
				rowStartIndex = _lastRequestOffset;
				rowEndIndex = _lastRequestOffset = Math.Min(_theBigTable.Rows.Count, _lastRequestOffset + Config.ChunkSize);
				
//				if (_theBigTable.Rows.Count == rowEndIndex)
//					TableWearhouse.Instance.Remove(_id);
			}
			if (rowStartIndex >= rowEndIndex)
				return null;

			for (int i = rowStartIndex; i < rowEndIndex; i++)
				chunk.Rows.Add(_theBigTable.Rows[i].ItemArray);	//	<-- faster
			//chunk.ImportRow(dr);			<-- slower

			Helper.TRACE("GetChunk {0} {1} - {2}", chunk.TableName, Helper.FormatLong(rowStartIndex), Helper.FormatLong((rowEndIndex)));


			return chunk;
		}

	}
	#endregion

	#region TableWearhouse

	/// <summary>
	/// This container contains all serviced tables.
	/// </summary>
	[ServiceBehavior(InstanceContextMode = InstanceContextMode.Single,
					 ConcurrencyMode = ConcurrencyMode.Multiple)]
	public class TableWearhouse : ITableTransporterService
	{
		#region members
		static Dictionary<long, Table> _tables = new Dictionary<long, Table>();
		static TableWearhouse _instance;
		static ServiceHost _wearhouseHost = null;
		int nextId = 0;
		static string _serviceBaseAddress;
		static string _serviceAddress;
		#endregion

		#region internal helpers
		internal static string URL
		{
			get
			{
				return _wearhouseHost.BaseAddresses[0].ToString();
			}
		}

		static public TableWearhouse Instance
		{
			get
			{
				if (_instance == null)
				{
					_instance = new TableWearhouse();
					StartService(6500);
				}
				return _instance;
			}
		}

		internal int Add(Table t)
		{
			int key = nextId++;
			lock (_tables)
				_tables.Add(key, t);
			
			return t.Id = key;
		}
		internal void Remove(int key)
		{
			lock (_tables)
			{
				if (_tables.ContainsKey(key))
				{
					Helper.TRACE("Transporting {0}  {1} [MB] took {2} [sec]", GetTable(key).TableName, Helper.FormatLong(_tables[key].SizeBytes/(1024*1024)), DateTime.Now.Subtract(_tables[key].ConstructionTime).TotalSeconds);
					_tables.Remove(key);
				}
			}
		}

		internal Table GetTable(int tableId)
		{
			lock (_tables)
				return _tables.ContainsKey(tableId) ? _tables[tableId] : null;
		}
		#endregion

		#region ITableTransporterService Members

		public DataTable NextChunk(int tableId)
		{
			try
			{
				if (Thread.CurrentThread.Name == null)
					Thread.CurrentThread.Name = "WCF ChunkTransporter Server";

				Table t = GetTable(tableId);

				if (t == null)
					return null;

				DataTable dt = t.GetChunk();

				if (dt == null)
					Remove(tableId);

				return dt;
			}
			catch (System.Exception e)
			{
				Helper.LogException("NextChunk", e);
				throw e;
			}
		}

		#endregion

		#region Host The Service
#if DEBUG
		static void AddMexEndpoint(ServiceHost host)
		{
			ServiceMetadataBehavior metadataBehavior = host.Description.Behaviors.Find<ServiceMetadataBehavior>();
			if (metadataBehavior == null)
			{
				metadataBehavior = new ServiceMetadataBehavior();
				metadataBehavior.HttpGetEnabled = true;

				host.Description.Behaviors.Add(metadataBehavior);
			}

			BindingElement bindingElement = new TcpTransportBindingElement();
			CustomBinding binding = new CustomBinding(bindingElement);
			host.AddServiceEndpoint(typeof(IMetadataExchange), binding, "MEX");
		}
#endif
		static public void StartService(int port)
		{

			if (_wearhouseHost != null)
				return;

			_serviceBaseAddress = string.Format("net.tcp://{0}:{1}", System.Net.Dns.GetHostName(), port);

			string myIp = System.Net.Dns.GetHostEntry(System.Net.Dns.GetHostName()).AddressList[0].ToString();
			_serviceBaseAddress = string.Format("net.tcp://{0}:{1}", myIp, port);
			_serviceAddress = string.Format("{0}/{1}", _serviceBaseAddress, "TableWearhouse");
			string httpAddress = string.Format("http://{0}:{1}/TableWearhouse", myIp, 8086);//port + 1);

			_wearhouseHost = new ServiceHost(new ChunkTransporter.TableWearhouse(),
				new System.Uri(_serviceAddress)
#if DEBUG
				,new System.Uri(httpAddress)
#endif
				);
			Binding binding;
#if DEBUG
			binding = new BasicHttpBinding();
			binding.CloseTimeout = binding.ReceiveTimeout = binding.SendTimeout = binding.OpenTimeout = TimeSpan.FromMinutes(10);
			(binding as BasicHttpBinding).MaxReceivedMessageSize = 100000000;
			_wearhouseHost.AddServiceEndpoint(
				typeof(ChunkTransporter.ITableTransporterService), binding, "");
#endif
			binding = new NetTcpBinding();
			
			binding.CloseTimeout = binding.ReceiveTimeout = binding.SendTimeout = binding.OpenTimeout = TimeSpan.FromMinutes(10);
			(binding as NetTcpBinding).MaxReceivedMessageSize = 100000000;
			_wearhouseHost.AddServiceEndpoint(
				typeof(ChunkTransporter.ITableTransporterService), binding, "");

#if DEBUG
			AddMexEndpoint(_wearhouseHost);
#endif

			//includeExceptionDetailInFaults
			_wearhouseHost.Description.Behaviors.Find<ServiceDebugBehavior>().IncludeExceptionDetailInFaults = true;

			try
			{
				// open the service
				_wearhouseHost.Open();
				Helper.TRACE("ChunkTransporter started {0}", _serviceAddress);
			}
			catch (System.ServiceModel.AddressAlreadyInUseException)
			{
				_wearhouseHost = null;
				// port taken, retry...
				StartService(port + 2);
			}
			catch (System.Exception e)
			{
				Helper.LogException("ServiceHost.Open()", e);
				throw;
			}
		}
		#endregion

	}
	#endregion
}